# 这是一个针对淘宝商品数据的ETL，主要完成：
#
# 数据抽取：从Excel文件读取原始数据
# 数据清洗：处理缺失值、去重、格式化字段
# 文本转换：提取商品标题关键词
# 数据聚合：按省份、店铺、价格区间生成统计指标
# 数据分层存储：按照数据仓库标准分为ODS、DWD、DWS三层


# 输出数据解析

# ODS层（原始数据）
# 文件：ods_raw_data.csv
# 内容：
# 原始Excel文件的完整数据
# 新增两个元数据字段：
# source_file：来源文件名
# load_time：数据加载时间
#
# DWD层（清洗后的明细数据）
# 文件：dwd_products.csv
# 价格字段提取数字（如从"¥59.9"转为59.9）
# 销量字段清洗（如从"1000+"转为1000）
# 包邮标志转为0/1（"包邮"->1）
# 地区拆分为省/市两列
# URL添加https前缀
# 文本特征表：dwd_product_text_features.csv
# 使用jieba分词提取标题关键词（如"男士短袖"->"男士 短袖"）
# 通过TF-IDF生成100维关键词特征向量（数值矩阵）
#
# DWS层（聚合数据）
# 文件：dws_province_sales.csv
# 统计指标：各省份销售总额、平均价格、商品数量、总销量、包邮率


# 如何使用这些文件- 数据分析：直接使用DWS层的聚合表进行可视化
# 模型训练：使用DWD层的文本特征表构建推荐系统
# 数据溯源：通过ODS层核查原始数据问题
# 业务监控：定期运行生成DWS表，监控销售趋势
# 如果需要调整任何处理逻辑（如修改价格区间范围），可以重点查看clean_data()和aggregate_data()方法

import os
import pandas as pd
import numpy as np
from datetime import datetime
import re
import jieba
from sklearn.feature_extraction.text import TfidfVectorizer
import warnings

warnings.filterwarnings('ignore')


class TaobaoETL:
    def __init__(self):
        # 创建输出目录
        os.makedirs('outputs/ods', exist_ok=True)
        os.makedirs('outputs/dwd', exist_ok=True)
        os.makedirs('outputs/dws', exist_ok=True)

        # 初始化分词器
        jieba.initialize()
        self.vectorizer = TfidfVectorizer(max_features=100)

    def extract(self):
        """抽取原始数据到ODS层"""
        all_data = []
        for file in os.listdir('data'):
            # 跳过临时文件和隐藏文件
            if file.startswith('~$') or not file.endswith('.xlsx'):
                continue

            try:
                file_path = os.path.join('data', file)
                df = pd.read_excel(file_path)

                # 打印列名用于调试
                print(f"文件 {file} 的列名:", df.columns.tolist())

                df['source_file'] = file
                df['load_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                all_data.append(df)
            except Exception as e:
                print(f"处理文件 {file} 时出错: {str(e)}")
                continue

        if not all_data:
            raise ValueError("未找到任何可用的XLSX文件")

        ods_df = pd.concat(all_data, ignore_index=True)

        # 保存到ODS层
        os.makedirs('outputs/ods', exist_ok=True)
        ods_df.to_csv('outputs/ods/ods_raw_data.csv', index=False, encoding='utf-8-sig')
        return ods_df

    def clean_data(self, df):
        """数据清洗模块"""
        # 检查必要列是否存在
        required_cols = ['title', 'Price', 'Deal', 'Location', 'Shop', 'IsPostFree']
        missing_cols = [col for col in required_cols if col not in df.columns]
        if missing_cols:
            raise ValueError(f"数据缺少必要列: {missing_cols}")

        # 去重（基于title+Shop组合）
        df = df.drop_duplicates(subset=['title', 'Shop'], keep='last')

        # 处理价格字段
        df['price'] = df['Price'].astype(str).str.extract(r'(\d+\.?\d*)')[0].astype(float)

        # 处理销量字段（将"5000+"转为5000）
        df['deal_num'] = df['Deal'].astype(str).str.replace(r'\+|\D', '', regex=True)
        df['deal_num'] = pd.to_numeric(df['deal_num'], errors='coerce').fillna(0).astype(int)

        # 处理包邮字段
        df['is_free_shipping'] = df['IsPostFree'].apply(
            lambda x: 1 if str(x) == '包邮' else 0
        )

        # 拆分地区信息
        location_split = df['Location'].str.extract(r'(\S+)\s+(\S+)')
        df['province'] = location_split[0]
        df['city'] = location_split[1]

        # URL处理
        df['product_url'] = df['Title_URL'].apply(
            lambda x: f"https:{x}" if str(x).startswith('//') else x
        )

        # 图片URL处理
        df['img_url'] = df['Img_URL'].apply(
            lambda x: f"https:{x}" if str(x).startswith('//') else x
        )

        return df

    def transform_text(self, df):
        """文本转换模块"""

        # 标题关键词提取
        def extract_keywords(text):
            words = [word for word in jieba.cut(str(text)) if len(word) > 1 and not word.isdigit()]
            return ' '.join(words)

        df['title_keywords'] = df['title'].apply(extract_keywords)

        # 生成TF-IDF特征（示例）
        tfidf_matrix = self.vectorizer.fit_transform(df['title_keywords'])
        keywords_df = pd.DataFrame(
            tfidf_matrix.toarray(),
            columns=[f"kw_{i}" for i in range(tfidf_matrix.shape[1])]
        )

        return pd.concat([df, keywords_df], axis=1)

    def aggregate_data(self, df):
        """数据聚合模块"""
        # 按省份聚合销售数据
        province_sales = df.groupby('province').agg({
            'price': ['sum', 'mean', 'count'],
            'deal_num': 'sum',
            'is_free_shipping': 'mean'
        }).reset_index()

        province_sales.columns = [
            'province', 'total_sales', 'avg_price', 'product_count',
            'total_deals', 'free_shipping_rate'
        ]

        # 按店铺聚合
        shop_stats = df.groupby('Shop').agg({
            'price': ['mean', 'count'],
            'deal_num': 'sum',
            'title': lambda x: x.nunique()
        }).reset_index()

        shop_stats.columns = [
            'shop', 'avg_price', 'product_count',
            'total_deals', 'unique_title_count'
        ]

        # 价格区间分析
        df['price_range'] = pd.cut(df['price'], bins=[0, 50, 100, 200, np.inf],
                                   labels=['<50', '50-100', '100-200', '>200'])

        price_dist = df.groupby('price_range').agg({
            'title': 'count',
            'deal_num': 'sum'
        }).reset_index()

        return province_sales, shop_stats, price_dist

    def load_to_dwd(self, df):
        """加载到DWD层"""
        # 商品明细表
        product_cols = [
            'title', 'price', 'deal_num', 'province', 'city',
            'Shop', 'is_free_shipping', 'product_url', 'img_url'
        ]
        df[product_cols].to_csv('outputs/dwd/dwd_products.csv', index=False, encoding='utf-8-sig')

        # 文本特征表
        text_cols = ['title', 'title_keywords'] + [c for c in df.columns if c.startswith('kw_')]
        df[text_cols].to_csv('outputs/dwd/dwd_product_text_features.csv', index=False, encoding='utf-8-sig')

    def load_to_dws(self, dfs):
        """加载到DWS层"""
        # 省份销售聚合表
        dfs[0].to_csv('outputs/dws/dws_province_sales.csv', index=False, encoding='utf-8-sig')

        # 店铺统计表
        dfs[1].to_csv('outputs/dws/dws_shop_stats.csv', index=False, encoding='utf-8-sig')

        # 价格分布表
        dfs[2].to_csv('outputs/dws/dws_price_distribution.csv', index=False, encoding='utf-8-sig')

    def run(self):
        """执行完整ETL流程"""
        print("1. 抽取数据到ODS层...")
        raw_df = self.extract()

        print("2. 数据清洗...")
        cleaned_df = self.clean_data(raw_df)

        print("3. 文本转换...")
        transformed_df = self.transform_text(cleaned_df)

        print("4. 数据聚合...")
        province_sales, shop_stats, price_dist = self.aggregate_data(transformed_df)

        print("5. 加载到DWD层...")
        self.load_to_dwd(transformed_df)

        print("6. 加载到DWS层...")
        self.load_to_dws([province_sales, shop_stats, price_dist])

        print("ETL流程完成！")


if __name__ == "__main__":
    etl = TaobaoETL()
    etl.run()